<?php

/**
 * 队列主入口
 *
 * @author suning <snsnsky@gmail.com>
 *
 */

declare(ticks = 1);
require __DIR__ . "/lib/config.php";

/**
 *
 * 队列Queue抽象类
 *
 */
abstract class Queue
{
    /**
     * 队列类
     *
     * @var QueueRedis
     */
    public $queue = null;

    /**
     * 最后一次的数据
     *
     * @var mixed
     */
    public $__lastdata;

    /**
     * 类名
     *
     * @var string
     */
    public $__name;

    /**
     * 是否包含未处理数据
     *
     * @var bool
     */
    public $__hascache = false;

    /**
     *  用于处理队列返回的数据
     *
     * @param string $data          队列中存入的数据
     */
    abstract public function run($data);

    /**
     * 返回PDO的连接, *结束时注意手工关闭*
     *
     * @param string $dbname db名称
     * @return PDO
     */
    public function getPDO($dbname = 'meite_admin')
    {
        global $cfg;
        $db_cfg = $cfg['mysql']['default']['w'];

        $dsn = "mysql:dbname={$dbname};host={$db_cfg['host']};charset=utf8";

        $pdo = new PDO($dsn, $db_cfg['user'], $db_cfg['password']);
        $pdo->setAttribute(PDO::ATTR_ERRMODE, PDO::ERRMODE_WARNING);

        return $pdo;
    }

    public function getReadPDO($dbname = 'meite_admin')
    {
        global $cfg;
        $db_cfg = $cfg['mysql']['default']['r'];

        $dsn = "mysql:dbname={$dbname};host={$db_cfg['host']};charset=utf8";

        $pdo = new PDO($dsn, $db_cfg['user'], $db_cfg['password']);
        $pdo->setAttribute(PDO::ATTR_ERRMODE, PDO::ERRMODE_WARNING);

        return $pdo;
    }

    /**
     * 返回Memcached连接, *结束时注意手工关闭*
     *
     * @param string $group MC分组名称
     * @return Memcached
     */
    public function getMemcache($group = 'default')
    {
        global $cfg;
        $mc = new \Memcached();
        $host = [];
        foreach ($cfg['memcache'][$group] as $v) {
            $host[] = ['host' => $v['host'], 'port' => $v['port']];
        }
        $mc->setOption(\Memcached::OPT_DISTRIBUTION, \Memcached::DISTRIBUTION_CONSISTENT);
        $mc->addServers($host);

        return $mc;
    }

    /**
     * 返回Redis连接, *结束时注意手工关闭*
     *
     * @param string $group Redis分组名称
     * @param int    $db    数据库位置
     * @return Redis
     */
    public function getRedis($group = 'other', $db = 0)
    {
        global $cfg;
        $conf = $cfg['redis'][$group]['w'];

        $redis = new \Redis();
        $redis->connect($conf['host'], $conf['port']);
        $redis->select($db);

        return $redis;
    }

    /**
     * 获取ES实例
     *
     * @return ElasticSearch
     */
    public function getES()
    {
        static $es = null;
        if ($es === null) {
            global $cfg;
            $config = $cfg['search']['default'];
            $es = new ElasticSearch($config);
        }

        return $es;
    }

    /**
     * 获取队列[运行时], 间隔频率, 单位 微秒
     */
    public function getUSleepTime()
    {
        return 500000;
    }

    /**
     * 获取队列[空闲时], 释放资源时间, 单位 秒
     */
    public function getSleepTime()
    {
        return 5;
    }

    /**
     * 获取队列运行时间段, 单位 秒
     */
    public function getRunPeriod()
    {
        return null;
    }

    /**
     * 队列未取到数据时的活动信号传递
     *
     * @return boolean
     */
    public function activeSignal()
    {
        return true;
    }

    /**
     * 队列被终止时调用方法
     *
     * @return boolean
     */
    public function shutDownSignal()
    {
        $this->log('QueueKillInfo', array('Killed', $this->__name, date("Y-m-d H:i:s"), json_encode($this->__lastdata)));
        exit;
    }

    /**
     * 记录日志
     *
     * @param string       $classname 日志名称, 一般为__CLASS__
     * @param array|string $data      日志数据
     * @param string       $sep       分隔符
     * @return bool
     */
    public function log($classname, $data, $sep = "\t")
    {
        $filepath = QUEUE_LOG . $classname . '_' . date("Ymd") . '.log';
        if (!is_array($data)) {
            $data = [$data];
        }
        foreach ($data as $k => &$v) {
            if (is_array($v) || is_object($v)) {
                $v = json_encode($v, JSON_UNESCAPED_UNICODE);
            }
            if (!is_numeric($k)) {
                $v = $k . ':' . $v;
            }
        }
        $data = implode($sep, $data) . "\n";
        $fp = fopen($filepath, 'a+');
        if ($fp) {
            flock($fp, LOCK_EX);
            fwrite($fp, $data);
            flock($fp, LOCK_UN);
            fclose($fp);

            return true;
        }

        return false;
    }

    /**
     *  跟踪调试
     *
     * @param String $error_info 错误内容
     * @return void
     */
    public function errorTrace($error_info)
    {
        ob_start();
        debug_print_backtrace();
        $trace = ob_get_contents();
        ob_end_clean();

        $data = array(
            "info" => $error_info,
            $trace
        );

        $this->log('Debug', $data, "\n");
    }
}

/**
 * Class QueueRedis
 */
class QueueRedis
{
    /**
     * Redis 实例
     *
     * @var \Redis
     */
    private $redis;

    public function __construct()
    {
        global $cfg;
        $conf = $cfg['redis']['queue']['w'];
        $this->redis = new \Redis();
        $this->redis->pconnect($conf['host'], $conf['port']);
        $this->redis->select(0);
    }

    /**
     * 获取队列值
     *
     * @param string $name 队列名
     * @return bool|string
     */
    public function get($name)
    {
        $data = $this->redis->lPop($name);
        if (empty($data)) {
            return false;
        }

        return $data;
    }

    /**
     * 设置队列值
     *
     * @param string $name 队列名
     * @param string $data 队列数据
     * @return int
     */
    public function add($name, $data)
    {
        $data = is_string($data) ? trim($data) : serialize($data);

        return $this->redis->rPush($name, $data);
    }

    /**
     * 向队列中添加多条数据
     *
     * @param string $name 队列名
     * @param array  $data 队列数据数组
     * @return boolean
     */
    public function addMulti($name, array $data)
    {
        if (empty($name) || empty($data)) {
            return false;
        }

        $pured = [];
        $result = false;

        foreach ($data as $item) {
            $pured[] = is_string($item) ? trim($item) : serialize($item);
        }

        $chunks = array_chunk($pured, 100);

        foreach ($chunks as $chunk) {
            array_unshift($chunk, $name);

            $result |= call_user_func_array([$this->redis, 'rPush'], $chunk);
        }

        if (!$result) {
            return false;
        }

        return true;
    }
}

/**
 *
 * 队列QueueRun 主程序
 *
 */
class QueueRun
{
    /**
     * 队列名称
     *
     * @var String
     */
    private $name;

    /**
     * 与$name对应的模块对象
     *
     * @var Queue
     */
    private $model;

    /**
     * 队列基类
     *
     * @var QueueRedis
     */
    private $queue;

    /**
     * 队列空闲Sleep时间
     *
     * @var integer
     */
    private $sleeptime;

    /**
     * 队列运行间隔时间
     *
     * @var integer
     */
    private $usleeptime;

    /**
     * 队列运行时间段
     *
     * @var array
     */
    private $run_period;

    /**
     * 构造函数
     *
     * @param string $name 队列名称
     *
     */
    public function __construct($name)
    {
        $model_file = QUEUE_ROOT . 'class/' . $name . '.class.php';

        if (!file_exists($model_file)) {
            exit("Queue '{$name}' not found!\n");
        }

        require $model_file;

        $this->name = $name;
        $this->model = new $name();
        $this->model->queue = $this->queue = new QueueRedis();

        $this->sleeptime = intval($this->model->getSleepTime());
        $this->usleeptime = intval($this->model->getUSleepTime());
        $this->run_period = $this->model->getRunPeriod();

        if (function_exists('pcntl_signal')) {
            $this->model->__name = $name;
            pcntl_signal(SIGTERM, array($this->model, 'shutDownSignal'));
            pcntl_signal(SIGINT, array($this->model, 'shutDownSignal'));
        }
    }

    /**
     * 队列循环入口
     *
     */
    public function doLoop()
    {
        $maxNotUsedCount = 6;
        $count = 0;
        while (true) {
            // 判断运行时段
            if (!empty($this->run_period)) {
                $now = time();
                $seconds = $now % 86400;

                // 当前时间已超出运行时段, 休眠至下一个运行开始时段
                if ($seconds >= $this->run_period[1]) {
                    $resume_time = ($now - $seconds) + 86400 + $this->run_period[0];

                    time_sleep_until($resume_time);
                }
            }

            $data = $this->queue->get($this->name);

            if ($data !== false) {
                $count = 0;
                $this->model->__lastdata = $data;
                $this->model->run($data);
            } else {
                $this->model->activeSignal();
                $count = $count > $maxNotUsedCount ? $maxNotUsedCount : ++$count;
            }

            if ($count == $maxNotUsedCount) {
                if ($this->sleeptime > 0) {
                    sleep($this->sleeptime);
                }
            } else {
                if ($this->usleeptime > 0) {
                    usleep($this->usleeptime);
                }
            }
        }
    }
}

if (isset($argc) && $argc > 1) {
    $queue = new QueueRun($argv[1]);
    $queue->doLoop();
} else {
    echo "Param Error!\n";
    echo "Usage:php Queue.php QueueName &\n";
}